package cn.doitedu.api;


import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/2/28
 * @Desc: 学大数据，上多易教育
 *
 *    自己实现的inner join
 *     不管数据进入的时间有何差别，只要满足join条件的数据，都能成功关联
 **/
public class _32_Exercise_OurInnerJoin {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        // {"order_id":"o1","order_amount":3000}
        DataStream<OrderMain> stream1 = env.
                socketTextStream("doitedu", 8891)
                .map(s -> JSON.parseObject(s, OrderMain.class));

        // o1,p1,1200
        // o1,p2,1800
        DataStream<OrderItem> stream2 = env
                .socketTextStream("doitedu", 8892)
                .map(s -> JSON.parseObject(s, OrderItem.class));


        stream1.connect(stream2)
                .keyBy(OrderMain::getOrderId,OrderItem::getOrderId)
                .process(new KeyedCoProcessFunction<String, OrderMain, OrderItem, String>() {

                    ListState<OrderMain> leftState;
                    ListState<OrderItem> rightState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        leftState = getRuntimeContext().getListState(new ListStateDescriptor<OrderMain>("left", OrderMain.class));
                        rightState = getRuntimeContext().getListState(new ListStateDescriptor<OrderItem>("right", OrderItem.class));

                    }

                    // 处理左流数据
                    @Override
                    public void processElement1(OrderMain orderMain, KeyedCoProcessFunction<String, OrderMain, OrderItem, String>.Context ctx, Collector<String> out) throws Exception {

                        // 遍历右表状态，逐条拼接
                        for (OrderItem orderItem : rightState.get()) {
                            out.collect(orderMain+" -> " + orderItem);
                        }
                        // 然后将本次的数据，放入状态
                        leftState.add(orderMain);

                    }

                    // 处理右流数据
                    @Override
                    public void processElement2(OrderItem orderItem, KeyedCoProcessFunction<String, OrderMain, OrderItem, String>.Context ctx, Collector<String> out) throws Exception {

                        // 遍历左表状态，逐条拼接
                        for (OrderMain orderMain : leftState.get()) {
                            out.collect(orderMain + " -> " + orderItem);
                        }

                        // 然后将本次的数据，放入右表状态
                        rightState.add(orderItem);


                    }
                }).print();


        env.execute();


    }
}
